package com.it.source;

import com.it.pojo.Event;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Random;

/**
 * 自定义source类型：主要实现run和cancel方法.
 *
 * @author code1997
 */
public class ConsumerSource {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        executionEnvironment.setParallelism(1);

        DataStreamSource<Event> sourceStream = executionEnvironment.addSource(new SourceFunction<Event>() {
            private volatile boolean running = true;

            @Override
            public void run(SourceContext<Event> ctx) throws InterruptedException {
                //随机生成数据
                Random random = new Random();
                //定义字段的选取的数据集
                String[] users = {"tom", "ailis", "kangkang"};
                String[] urls = {"./index", "./home", "./fav", "./order?id=10"};
                //执行数据的范围
                while (running) {
                    String user = users[random.nextInt(users.length)];
                    String url = urls[random.nextInt(urls.length)];
                    ctx.collect(new Event(user, url, System.currentTimeMillis()));
                    Thread.sleep(100);
                }
            }

            @Override
            public void cancel() {
                //update flag to control job
                running = false;
            }
        });
        sourceStream.print();
        executionEnvironment.execute();
    }
}
